package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.PortInfo;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

/**
 * 异常端口检测Bolt
 * 发现常见木马端口以及常见开放风险端口
 * @author Administrator
 *
 */
public class PortDetectBolt extends AbstractBolt {

	/**
	  *
	  */
	    
	private static final long serialVersionUID = 3126846279638077446L;

	HashMap<Integer,PortInfo> ports = null;
	
	HashMap<Integer,PortInfo> portsScan = null;
	
	HTable _hTable = null; 
	
	HashMap<Long,IPDatabase> map = null;
	
	HashMap<String,Integer> _mapDIPandPort 
		= new HashMap<String,Integer>();
	
	HashMap<String,Integer> _mapPortScanStat 
		= new HashMap<String,Integer>();
	
	/**
	 * 端口扫描风险
	 */
	HashMap<String,List<String>> mapPortScanRisk 
		= new HashMap<String,List<String>>();
	
	HashMap<String,Integer> _mapPortTrojan
	= new HashMap<String,Integer>();
	
	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> _mapStreamoctets = new HashMap<String,Long>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Long> _mapStreampacket = new HashMap<String,Long>();
	
	
	/**
	 * 常用服务端口
	 */
	HashMap<String,HashMap<Integer,Integer>> _mapServicePort = 
			new HashMap<String,HashMap<Integer,Integer>>();
	
	
	//
	HashMap<String,String> ipMap = null; 
	
	public PortDetectBolt() {
		
	}
			
	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		arg0.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName",
				"pv","dateStr_hour","abntype", "idcno", "sourcecountry",
				"streamoctets","streampacket"));
		
		arg0.declareStream("STAT",new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName",
				"pv","dateStr_hour","abntype", "idcno", "sourcecountry",
				"streamoctets","streampacket"));
		
		arg0.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
		
		arg0.declareStream("DELAY_", new Fields("DIP","SIP","AREANAME","PV","SOURCECOUNTRY","STREAMOCTETS","STREAMPACKET","ABNTYPE"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 60);
		return conf;
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		
		
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				//emitCountingData();  
				
				/**加载前一天的端口服务配置数据,记录常用服务端口，其他端口为异常端口
				记录某个端口在五分钟内被访问的次数，设定阀值，如果不是前一天常用访问端口，提出告警
				//function**/
//				LoadServicePort();
				PortScanDetect();
				PortTrojanDetect();
				_mapPortScanStat.clear();
				_mapStreamoctets.clear();
				_mapStreampacket.clear();
				_mapPortTrojan.clear();
				mapPortScanRisk.clear();
				
				collector.ack(tuple);
			} else {
				//
				PortScanCount(tuple);
				//PortScanDetect(tuple);
				//countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
//			System.err.println(" PortDetectBolt exception"+e.getMessage());
			Logger.getLogger("STDERR").error("PortDetectBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException("PortDetectBolt exception", e);
		}	
	}
	
	private void PortScanCount(Tuple tuple)
	{
		try
		{
			String destip = tuple.getString(0);
			int destport = Integer.parseInt(tuple.getString(1));
			
			if(!ports.containsKey(destport)) {
				collector.ack(tuple);
				return;
			}
			
			PortInfo portinfo = ports.get(destport);
			String sourceip = tuple.getString(2);
			byte[] gisbytes =  Base64.decodeBase64(tuple.getStringByField("gisstr"));
			String gisstr = new String(gisbytes);
			String idc = tuple.getStringByField("idcno");
			
			
			//非常用端口，疑似木马开放端口检测
			String trojanKey = String.format("%s|%s|%d|%s|%s", 
					destip, sourceip, destport, gisstr, idc);
			if(portinfo.NormalType == 2) {
				int count = 1;
				if(_mapPortTrojan.containsKey(trojanKey))
					count = _mapPortTrojan.get(trojanKey) + 1;
				_mapPortTrojan.put(trojanKey, count);
				
				if(tuple.contains("upstreamoctets") && tuple.contains("upstreampacket")) {
					long nUpStreamOctets = tuple.getLongByField("upstreamoctets");
					long nUpstreampacket = tuple.getLongByField("upstreampacket");
					if(_mapStreamoctets.containsKey(trojanKey)) {
						nUpStreamOctets = _mapStreamoctets.get(trojanKey) + nUpStreamOctets;
					}
					_mapStreamoctets.put(trojanKey, nUpStreamOctets);
					
					if(_mapStreampacket.containsKey(trojanKey)) {
						nUpstreampacket = _mapStreampacket.get(trojanKey) + nUpstreampacket;
					}
					_mapStreampacket.put(trojanKey, nUpstreampacket);
				
				}
				
				collector.ack(tuple);
				return;
			}
			
			//-----异常日志入库
		
			String desc = "";
			int score = 100;			
			if(portinfo!=null) {
				desc = desc + "" +portinfo.Desc;
				score = portinfo.score;
			}
			
			String key = String.format("%s|%s|%d|%s|%s|%d|%s",	
					destip, sourceip, destport, gisstr, desc, score, idc);
			
			int count = 1;
			if(_mapPortScanStat.containsKey(key))
				count = _mapPortScanStat.get(key) + 1;
			_mapPortScanStat.put(key, count);
			
			
			//UDII数据特有
			if(tuple.contains("upstreamoctets") && tuple.contains("upstreampacket")) {
				long nUpStreamOctets = tuple.getLongByField("upstreamoctets");
				long nUpstreampacket = tuple.getLongByField("upstreampacket");
				if(_mapStreamoctets.containsKey(key)) {
					nUpStreamOctets = _mapStreamoctets.get(key) + nUpStreamOctets;
				}
				_mapStreamoctets.put(key, nUpStreamOctets);
				
				if(_mapStreampacket.containsKey(key)) {
					nUpstreampacket = _mapStreampacket.get(key) + nUpstreampacket;
				}
				_mapStreampacket.put(key, nUpstreampacket);
			
			}			
		} catch (Exception e) {
			// TODO Auto-generated catch block
//			System.err.println("PortDetectBolt emitCountingData exception"+e.getMessage());
			Logger.getLogger("STDERR").error("PortDetectBolt PortScanCount exception====",e);
			throw new FailedException("PortDetectBolt PortScanCount exception====", e);
		}
		
	}
	
	private void PortScanDetect() {
		try {
			for(Entry<String,Integer> entry : _mapPortScanStat.entrySet()) {
				
				String key = entry.getKey();
				String values[] = key.split("\\|");
				String destip = values[0];
				String sourceip = values[1];
				String destport = values[2];
				String gisstr = values[3];
				String desc = values[4];
				String score = values[5];
				String idc = values[6];
				
				String[] gisArray = gisstr.split("#");
				
				//Logger.getLogger(PortDetectBolt.class).info("LANG DEBUG="+gisstr);
				
				String destAreaName = gisArray[0]; //如果地市为空，精确到国家
				String DestGeo = gisArray[1];
				String sourceAreaName = gisArray[2];  //如果地市为空，精确到国家
				String SourceGeo = gisArray[3];
				String sourceAreaCountry = gisArray[4];
				String sourceProvinceID =  gisArray[5];
				String sourceProvinceName = gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
				String sourceAreaCityId = gisArray[7];
				String sourceAreaProvinceId = gisArray[8];
				
				HBaseRecordAdd addDestIP = HBaseRecordAdd.getInstance(zooserver);
				
				

				Date current = new Date();
				SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
				String strDate = sdf.format(current);
				
				String rowkey = String.format("%s_%s_%s_%s_%d", destip, strDate, 
						sourceip,destport,4);
				
				String cf = "cf";

		    	//##########COPY LOG
		    	String tbName2 = ApolloProperties.SDS_ABNORMAL_LOG2;
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTIP", destip);
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTPORT", String.valueOf(destport));
		    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEIP", sourceip);
		    	addDestIP.Add(tbName2, rowkey, cf,  "ACCESSTIME", strDate);
		    	addDestIP.Add(tbName2, rowkey, cf,  "ENDTIME", strDate);
		    	addDestIP.Add(tbName2, rowkey, cf, "ABRNORMAL", "4");
		    	addDestIP.Add(tbName2, rowkey, cf,  "EVALUATE", String.valueOf(score));
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESC", desc);
		    	addDestIP.Add(tbName2, rowkey, cf,  "ATTNUM", String.valueOf(entry.getValue()));
		    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEAREA", sourceAreaName.replace("市", ""));
		    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEGEO", String.format("%s",SourceGeo));
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTAREA", String.format("%s",destAreaName.replace("市", "")));
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTGEO", String.format("%s",DestGeo));
		    	addDestIP.Add(tbName2, rowkey, cf, "SOURCECOUNTRY", sourceAreaCountry);
		    	addDestIP.Add(tbName2, rowkey, "cf", "PROVINCE", sourceProvinceName);
		    	
		    	String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		    	String rowkey3 = String.format("%s_%s", strDate, sourceip);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "ACCESSTIME", strDate);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCEIP", sourceip);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",SourceGeo));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(entry.getValue()));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "AREAID", sourceProvinceID);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTIP", destip);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTPORT", String.valueOf(destport));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "TYPE", "4");
		    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		    	addDestIP.Add(tbName3, rowkey3, cf,  "DESTGEO", String.format("%s",DestGeo));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
		    	
		    	SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
				SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
				SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		    	String dateStr_h = sdf_h.format(current);
				String dateStr_d = sdf_d.format(current);
				String dateStr_hour = sdf_hour.format(current);
				
				//UDII数据特有
				long nUpStreamOctets = 0;
				long nUpstreampacket = 0;
				if(_mapStreamoctets.size() > 0 && _mapStreampacket.size() > 0) {
					nUpStreamOctets = _mapStreamoctets.get(entry.getKey());
					nUpstreampacket = _mapStreampacket.get(entry.getKey());
					addDestIP.Add("SDS_ABNORMAL_LOG2", rowkey, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
					addDestIP.Add("SDS_ABNORMAL_LOG2", rowkey, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
				}
				
		    	collector.emit(new Values(destip,sourceip,String.valueOf(destport), "",
		    			dateStr_d,dateStr_h,sourceProvinceID,sourceAreaName,
		    			sourceProvinceName, 1l, dateStr_hour,"PORTLOW", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
		    	
		    	
		    	collector.emit("STAT",new Values(destip,sourceip,String.valueOf(destport), "",
		    			dateStr_d,dateStr_h,sourceProvinceID,sourceAreaName,
		    			sourceProvinceName, 1l, dateStr_hour,"PORTLOW", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
		   	
		    	collector.emit("DELAY_",new Values(destip,sourceip,sourceAreaName, (long)entry.getValue(),
		    			sourceAreaCountry, (long)nUpStreamOctets, (long)nUpstreampacket, "1"));
			}
			
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
//			System.err.println(" PortDetectBolt exception"+e.getMessage());
			Logger.getLogger("STDERR").error("PortDetectBolt exception====",e);
			throw new FailedException(e);
		}
	}
	

	private void PortTrojanDetect()	{
		try {
	//			String trojanKey = String.format("%s|%s|%d|%s|%s", 
	//					destip, sourceip, destport, gisstr, idc);
				for(Entry<String,Integer> entry : _mapPortTrojan.entrySet()) {
					if(entry.getValue() < 5*3) //小于每分钟3次
						continue;
					
					String key = entry.getKey();
					String values[] = key.split("\\|");
					String destip = values[0];
					String sourceip = values[1];
					String destport = values[2];
					String gisstr = values[3];
					String idc = values[4];
					
					String[] gisArray = gisstr.split("#");
					
					String destAreaName = gisArray[0]; //如果地市为空，精确到国家
					String DestGeo = gisArray[1];
					String sourceAreaName = gisArray[2];  //如果地市为空，精确到国家
					String SourceGeo = gisArray[3];
					String sourceAreaCountry = gisArray[4];
					String sourceProvinceID =  gisArray[5];
					String sourceProvinceName = gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
					String sourceAreaCityId = gisArray[7];
//					String sourceAreaProvinceId = gisArray[8];
					
					
					
					HBaseRecordAdd addDestIP = HBaseRecordAdd.getInstance(zooserver);
					
					PortInfo portinfo = ports.get(Integer.parseInt(destport));
					String desc = String.format("疑似 %s ,5分钟内有 %d 次访问", portinfo.Desc,entry.getValue());
					
	
					Date current = new Date();
					SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
					String strDate = sdf.format(current);
					
					String rowkey = String.format("%s_%s_%s_%s_%d", destip, strDate, 
							sourceip,destport,4);
					
					String cf = "cf";
	
			    	//##########COPY LOG
			    	String tbName2 = ApolloProperties.SDS_ABNORMAL_LOG2;
			    	addDestIP.Add(tbName2, rowkey, cf,  "DESTIP", destip);
			    	addDestIP.Add(tbName2, rowkey, cf,  "DESTPORT", String.valueOf(destport));
			    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEIP", sourceip);
			    	addDestIP.Add(tbName2, rowkey, cf,  "ACCESSTIME", strDate);
			    	addDestIP.Add(tbName2, rowkey, cf, "ABRNORMAL", "4");
			    	addDestIP.Add(tbName2, rowkey, cf,  "EVALUATE", String.valueOf(40));
			    	addDestIP.Add(tbName2, rowkey, cf,  "DESC", desc);
			    	addDestIP.Add(tbName2, rowkey, cf,  "ATTNUM", String.valueOf(entry.getValue()));
			    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEAREA", sourceAreaName.replace("市", ""));
			    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEGEO", String.format("%s",SourceGeo));
			    	addDestIP.Add(tbName2, rowkey, cf,  "DESTAREA", String.format("%s",destAreaName.replace("市", "")));
			    	addDestIP.Add(tbName2, rowkey, cf,  "DESTGEO", String.format("%s",DestGeo));
			    	addDestIP.Add(tbName2, rowkey, cf, "SOURCECOUNTRY", sourceAreaCountry);
			    	addDestIP.Add(tbName2, rowkey, "cf", "PROVINCE", sourceProvinceName);
			    	
			    	String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
			    	String rowkey3 = String.format("%s_%s", strDate, sourceip);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "ACCESSTIME", strDate);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCEIP", sourceip);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",SourceGeo));
			    	addDestIP.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(entry.getValue()));
			    	addDestIP.Add(tbName3, rowkey3, "cf", "AREAID", sourceProvinceID);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTIP", destip);
			    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTPORT", String.valueOf(destport));
			    	addDestIP.Add(tbName3, rowkey3, "cf", "TYPE", "4");
			    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
			    	addDestIP.Add(tbName3, rowkey3, cf,  "DESTGEO", String.format("%s",DestGeo));
			    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
			    	
			    	SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
					SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
					SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
			    	String dateStr_h = sdf_h.format(current);
					String dateStr_d = sdf_d.format(current);
					String dateStr_hour = sdf_hour.format(current);
					
					//UDII数据特有
					long nUpStreamOctets = 0;
					long nUpstreampacket = 0;
					if(_mapStreamoctets.size() > 0 && _mapStreampacket.size() > 0) {
						nUpStreamOctets = _mapStreamoctets.get(entry.getKey());
						nUpstreampacket = _mapStreampacket.get(entry.getKey());
						addDestIP.Add("SDS_ABNORMAL_LOG2", rowkey, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
						addDestIP.Add("SDS_ABNORMAL_LOG2", rowkey, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
					}
					
			    	collector.emit(new Values(destip,sourceip,String.valueOf(destport), "",
			    			dateStr_d,dateStr_h,sourceProvinceID,sourceAreaName,
			    			sourceProvinceName, 1l, dateStr_hour,"PORTLOW", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
			    	
			    	
			    	collector.emit("STAT",new Values(destip,sourceip,String.valueOf(destport), "",
			    			dateStr_d,dateStr_h,sourceProvinceID,sourceAreaName,
			    			sourceProvinceName, nUpStreamOctets, dateStr_hour,"PORTLOW", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
			    	
			    	collector.emit("DELAY_",new Values(destip,sourceip,sourceAreaName, entry.getValue(),
			    			sourceAreaCountry, (long)nUpStreamOctets, (long)nUpstreampacket, "1"));
				}
			} catch (Exception e) {
				// TODO Auto-generated catch block
//				System.err.println(" PortDetectBolt exception"+e.getMessage());
				Logger.getLogger("STDERR").error("PortDetectBolt exception====",e);
				throw new FailedException(e);
			}
	}
	
	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub

		Logger.getLogger(PortDetectBolt.class).info("#####storm.zookeeper.servers######" 
				+ stormConf.get("storm.zookeeper.servers").toString());
		//["192.168.5.95" "192.168.5.96" "192.168.5.97"]
		//String _zooServer = stormConf.get("storm.zookeeper.servers");
		
		String dbJson = stormConf.get("dbjson").toString();
		ApolloConfig ap = new ApolloConfig(dbJson);
		
		//加载配置信息
		ports = ap.getPortThreaten();
		
		ports.put(22,new PortInfo(22, 80, "SSH远程协议访问",0));
		ports.put(21,new PortInfo(21, 80, "ftp访问",0));
		ports.put(23,new PortInfo(23, 80, "Telnet访问",0));
		ports.put(3389, new PortInfo(3389, 80, "Windows远程终端访问",0));
		//135、139、445、593、1025
		ports.put(135, new PortInfo(135, 80, "Windows DCOM服务访问",0));
		ports.put(139, new PortInfo(139, 80, "NetBIOS/SMB服务访问",0));
		ports.put(445, new PortInfo(445, 80, "公共Internet文件系统访问",0));
		ports.put(1433, new PortInfo(1433, 80, "SQL Server常用端口访问",0));
	}
}
